﻿#include "MyTcpSocket.h"

#include <QtEndian>
#include <QHostAddress>

extern QUEUE_DATA<MESG> g_sendQueue;
extern QUEUE_DATA<MESG> g_recvQueue;
extern QUEUE_DATA<MESG> g_recvAudioQueue; //音频接收队列

MyTcpSocket::MyTcpSocket(QObject *parent):
    QThread(parent),
    m_sendBuff(new uchar[4 * MB]),
    m_recvBuff(new uchar[4 * MB]),
    m_recvBuffBytes(0),
    m_isRun(false)
{
    qRegisterMetaType<QAbstractSocket::SocketError>();

    m_tcpSocket = nullptr;

    m_tcpSocketThread = new QThread();
    this->moveToThread(m_tcpSocketThread);
    connect(m_tcpSocketThread, &QThread::finished, this, &MyTcpSocket::closeSocket);
}

MyTcpSocket::~MyTcpSocket()
{
    if (m_tcpSocket != nullptr) {
        m_tcpSocket->deleteLater();
    }

    delete [] m_sendBuff;
    delete [] m_recvBuff;

    m_tcpSocketThread->deleteLater();
}

bool MyTcpSocket::connectToServer()
{
    m_tcpSocketThread->start();

    bool retVal = true;

    // 使用异步函数调用另一个connectServer，则是在另一个线程（m_tcpSocketThread）中创建套接字连接服务器，同时在这个线程中和服务器通信
    QMetaObject::invokeMethod(this, "connectServer", Qt::BlockingQueuedConnection, Q_RETURN_ARG(bool, retVal));

    // 如何在另一个线程中连接服务器成功，则启动本线程run函数，不断从队列中取出msg，异步调用send函数
    if (retVal == true) {
        this->start();
    }
    return retVal;
}

QString MyTcpSocket::errorString()
{
    return m_tcpSocket->errorString();
}

void MyTcpSocket::disconnectFromHost()
{
    if (this->isRunning()) {
        QMutexLocker locker(&m_mutex);
        m_isRun = false;
    }

    if (m_tcpSocketThread->isRunning()) {
        m_tcpSocketThread->quit();
        m_tcpSocketThread->wait();
    }

    g_sendQueue.clear();
    g_recvQueue.clear();
    g_recvAudioQueue.clear();
}

bool MyTcpSocket::connectServer()
{
    if (nullptr == m_tcpSocket) {
        m_tcpSocket = new QTcpSocket();
    }

    QString ip{"121.199.21.233"};
    quint16 port{8888};
    m_tcpSocket->connectToHost(ip, port, QIODevice::ReadWrite);

    connect(m_tcpSocket, &QTcpSocket::readyRead, this, &MyTcpSocket::recvToSocket);
    connect(m_tcpSocket, SIGNAL(error(QAbstractSocket::SocketError)),
            this, SLOT(errorDetect(QAbstractSocket::SocketError)),Qt::UniqueConnection);

    if (m_tcpSocket->waitForConnected(5000)) {
        return true;
    }
    m_tcpSocket->close();
    return false;

}

quint32 MyTcpSocket::getLocalIp()
{
    if (m_tcpSocket->isOpen()) {
        return m_tcpSocket->localAddress().toIPv4Address();
    } else {
        return -1;
    }
}

void MyTcpSocket::StopTcpThread()
{
    {
        QMutexLocker locker(&m_mutex);
        m_isRun = false;
    }
    m_tcpSocketThread->quit();
    m_tcpSocketThread->wait();
}

void MyTcpSocket::recvToSocket()
{
    qint64 availableBytes = m_tcpSocket->bytesAvailable();
    LOG << "recv() availableBytes" << availableBytes << "id" << QThread::currentThreadId();
    if (availableBytes <= 0) {
        LOG << "无接收到有效数据"; return;
    }

    qint64 readBytes = m_tcpSocket->read((char*)m_recvBuff + 0, availableBytes);
    LOG << "recv() readBytes" << readBytes << "id" << QThread::currentThreadId();
    if (readBytes <= 0) {
        LOG << "读取错误/未读取到数据 readBytes =" << readBytes; return;
    }

    m_recvBuffBytes += readBytes;
    if (m_recvBuffBytes < MESG_HEAD_SIZE) {
        LOG << "接收缓冲区数据小于消息头"; return;
    }

    quint32 msgSize = 0;
    qFromBigEndian<quint32>(m_recvBuff + 7, MESG_SIZE_VAR_SIZE, &msgSize);
    LOG << "msg size " << msgSize;

    if (m_recvBuffBytes < MESG_HEAD_SIZE + msgSize + 1) {
        LOG << "接收缓冲区数据不足一个消息包"; return;
    }

    if (m_recvBuff[0] != '$') {
        LOG << "首位错误"; return;
    }

    quint16 type;
    qFromBigEndian<quint16>(m_recvBuff + 1, MESG_TYPE_VAR_SIZE, &type);
    MsgType msgType = (MsgType) type;
    LOG << "msgType " << msgType;

    quint32 msgIp;
    qFromBigEndian<quint16>(m_recvBuff + 1  + MESG_TYPE_VAR_SIZE, MESG_IP_VAR_SIZE, &msgIp);
    LOG << "msgIp " << msgType;

    MESG *msg = new (std::nothrow) MESG;
    if (nullptr == msg) {
        LOG << "msg 内存分配失败"; return;
    }

    msg->type = msgType;
    msg->size = msgSize;
    msg->ip   = msgIp;
    msg->data = nullptr;

    if (msgType == MsgType_CreatMeetingResponse) {
        qint32 roomNo;
        qFromBigEndian<qint32>(m_recvBuff + MESG_HEAD_SIZE, 4, &roomNo);
        LOG << "roomNo " << roomNo;

        if (msg->size > 0) {
            msg->data = std::shared_ptr<uchar[]>(new uchar[msg->size]);            // *异常处理*
        }
        memcpy(msg->data.get(), &roomNo, msg->size);
        g_recvQueue.push_msg(msg);

    } else if (msgType == MsgType_JoinMeetingResponse) {
        qint32 c;
        memcpy(&c, m_recvBuff + MESG_HEAD_SIZE, msg->size);

        if (msg->size > 0) {
            msg->data = std::shared_ptr<uchar[]>(new uchar[msg->size]);            // *异常处理*
        }
        memset(msg->data.get(), 0, msg->size);
        memcpy(msg->data.get(), &c, msg->size);
        g_recvQueue.push_msg(msg);

    } else if (msgType == MsgType_AudioRecv) {
        QByteArray cc((char*)m_recvBuff + MESG_HEAD_SIZE, msg->size);
        QByteArray rc = QByteArray::fromBase64(cc);
        QByteArray rdc = qUncompress(rc);
        if (rdc.size() > 0) {
            LOG << "Audio msg size <= 0";

            msg->size = rdc.size();

            msg->data = std::shared_ptr<uchar[]>(new uchar[rdc.size()]);
            memset(msg->data.get(), 0, rdc.size());
            memcpy_s(msg->data.get(), msg->size, rdc.data(), rdc.size());

            g_recvAudioQueue.push_msg(msg);
        }

    } else if (msgType == MsgType_ImgRecv){
        QByteArray cc((char*)m_recvBuff + MESG_HEAD_SIZE, msg->size);
        QByteArray rc = QByteArray::fromBase64(cc);
        QByteArray rdc = qUncompress(rc);
        if (rdc.size() > 0) {
            LOG << "Img msg size <= 0";

            msg->size = rdc.size();

            msg->data = std::shared_ptr<uchar[]>(new uchar[rdc.size()]);
            memset(msg->data.get(), 0, rdc.size());
            memcpy_s(msg->data.get(), rdc.size(), rdc.data(), rdc.size());

            g_recvQueue.push_msg(msg);
        }

    } else if (msgType == MsgType_PartnerJoin || msgType == MsgType_PartnerExit || msgType == MsgType_CloseCamera) {
        g_recvQueue.push_msg(msg);

    }

    memmove_s(m_recvBuff, 4 * MB,
              m_recvBuff + (MESG_HEAD_SIZE + msgSize + 1),
              m_recvBuffBytes - (MESG_HEAD_SIZE + msgSize + 1));
    m_recvBuffBytes -= (MESG_HEAD_SIZE + msgSize + 1);
}

void MyTcpSocket::run() {
    m_isRun = true;
    while (true) {
        {
            QMutexLocker locker(&m_mutex);      // 主线程会使用到m_isRun,所以需要加锁
            if (!m_isRun) {
                m_tcpSocket->close();
                return;
            }
        }
        auto msg = g_sendQueue.pop_msg();
        if (msg == nullptr) {
            continue;
        }

        QMetaObject::invokeMethod(this, "sendData", Q_ARG(MESG*, msg));
    }
}


void MyTcpSocket::sendData(MESG *msg)
{
    if (m_tcpSocket->state() == QAbstractSocket::UnconnectedState) {
        delete msg;
        return;
    }
    int sendBuffIndex = 0;
    m_sendBuff[sendBuffIndex++] = '$';                                        // 头部标识位

    qToBigEndian<quint16>(msg->type, m_sendBuff + sendBuffIndex);             // 填入数据类型
    sendBuffIndex += 2;

    quint32 ip = m_tcpSocket->localAddress().toIPv4Address();
    qToBigEndian<quint32>(ip, m_sendBuff + sendBuffIndex);                    // 填入发送者IP
    sendBuffIndex += 4;

    qToBigEndian<quint32>(msg->size, m_sendBuff + sendBuffIndex);             // 填入数据长度
    sendBuffIndex += 4;

    if (nullptr != msg->data) {
        memcpy(m_sendBuff + sendBuffIndex, msg->data.get(), msg->size);       // 填入需发送数据
        sendBuffIndex += msg->size;
    }

    m_sendBuff[sendBuffIndex++] = '#';                                        // 尾部标识位

    // 将缓冲区中字节通过 write() 发送
    qint64 bytesWaitWrite = sendBuffIndex, bytesWritten = 0;
    while (bytesWritten < sendBuffIndex ) {
        auto onceBytesWritten = m_tcpSocket->write((char*)m_sendBuff + bytesWritten, bytesWaitWrite);

        if (onceBytesWritten == -1) {
            if (m_tcpSocket->error() == QAbstractSocket::TemporaryError) {
                QThread::msleep(100);   //-*-
                continue;
            } else {
                qDebug() << "network error";//-*-
                break;
            }

        }
        bytesWritten += onceBytesWritten;
        bytesWaitWrite -= onceBytesWritten;
    }
    m_tcpSocket->waitForBytesWritten();

    delete msg;
}

void MyTcpSocket::errorDetect(QAbstractSocket::SocketError error)
{
   switch (error)
   {
   case QAbstractSocket::ConnectionRefusedError:
       qDebug() << "服务器拒绝连接";
       break;
   case QAbstractSocket::RemoteHostClosedError:
       qDebug() << "远程主机关闭连接";
       break;
   case QAbstractSocket::HostNotFoundError:
       qDebug() << "无法找到主机";
       break;
   // 其他错误情况的处理类似
   default:
       qDebug() << "其他错误：" << error;
   }
}


void MyTcpSocket::closeSocket()
{
    if (m_tcpSocket && m_tcpSocket->isOpen()) {
        m_tcpSocket->close();
    }
}

